package com.amon.amonplayer.rxjava;

import com.amon.amonplayer.rxjava.function.Function;

public class ObservableMap<T, R> extends Observable<R> {
    final ObservableSource<T> source;
    final Function<? super T, ? extends R> mapper;

    public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends R> mapper) {
        this.source = source;
        this.mapper = mapper;
    }

    @Override
    protected void subscribeActual(Observer<R> s) {
        source.subscribe(new MapObserver<>(s, mapper));
    }

    private static class MapObserver<T, R> implements Observer<T> {
        Observer<? super R> source;
        Function<? super T, ? extends R> mapper;

        public MapObserver(Observer<R> s, Function<? super T, ? extends R> mapper) {
            this.source = s;
            this.mapper = mapper;
        }

        @Override
        public void onNext(T t) {
            try {
                source.onNext(mapper.apply(t));
            } catch (Exception e) {
                onError(e);
            }
        }

        @Override
        public void onError(Throwable e) {
            source.onError(e);
        }

        @Override
        public void onComplete() {
            source.onComplete();
        }

        @Override
        public void onSubscribe() {
            source.onSubscribe();
        }
    }
}
